package com.leiyuee.flink.batch.job;

import cn.hutool.core.util.StrUtil;
import com.leiyuee.flink.tools.StatementUtil;
import com.leiyuee.flink.tools.constants.MySQLConstant;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.types.RowKind;
import org.apache.poi.hssf.usermodel.HSSFDateUtil;
import org.apache.poi.ss.usermodel.*;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.joda.time.DateTime;
import com.leiyuee.flink.tools.SinkBuilder;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * 1、读取目标路径Sheet：P1P2_For PBI 中的全部数据，注意：文件内容是每周增量添加，空一行空白行是每周的分割线
 * 2、读取频率：每周二9点开始，每小时读取一次
 * 3、读取到文件数据后，删除数据库的记录，重新插入数据
 * 4、文件导入失败时删除当前批次已导入的数据后重新导入"
 * 5. 读取 A列为1 的数据
 *
 * @author Yu_Lei
 * @date 2024/7/5
 */
public class ExcelP1ToKpiJob {
    public static void main(String[] args) throws Exception {
        System.out.println("ExcelP1ToKpiJob");
        String filePath = "/usr/local/app/cnhuait122/P1P2IncAnalysis/EMS_Weekly P1P2 Incident Analysis_2024.xlsx";

//        String filePath = "SQLscripts/EMS_Weekly P1P2 Incident Analysis_2024.xlsx" ;
        List<P1Entity> p1Entities = readExcel(filePath);
        if (p1Entities.size() == 0) {
            System.out.println("No data found in the Excel file.");
            return;
        }
        List<RowData> rowData = listToRowDataConverter(p1Entities);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String mysqlDorisDB = "Trace_KPI";
        String mysqlDorisTB = "tb_p1_history_data";
        String mysqlDorisURL = MySQLConstant.DORIS_PRD02_URL + mysqlDorisDB;

        // 1.清空表
        String truncateTableSQL = "TRUNCATE TABLE " + mysqlDorisTB;
        try (Connection connection = DriverManager.getConnection(mysqlDorisURL,  MySQLConstant.DORIS_PRD02_USER_JABIL, MySQLConstant.DORIS_PRD02_USER_JABIL_PWD);
             Statement statement = connection.createStatement()) {
            statement.executeUpdate(truncateTableSQL);
            System.out.println("Table truncated successfully!");
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            e.printStackTrace();
        }
        //2. 写数据
        // 读取数据
        DataStreamSource<RowData> dataStream = env.fromCollection(rowData);
        dataStream.print("RowData: \n");
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
        //         Doris 配置 Sink To Doris: http://cnhuam0itprd02/8030

        String mysqlDorisStr = String.format("INSERT INTO %s.%s ", mysqlDorisDB, mysqlDorisTB) +
                "(priority, location, `number`, issue_description, reason_category, opened, closed, mitunes, week, has_breached, mttr, emergency_change, prb_number, prb_status, actual_rc_found, configuration_item, impact, production_line_dt, action_taken, root_cause, problem_solution)\n" +
                "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?, ?);";
        String indexDorisField = String.format("[%s]%s.%s %s", "Doris MySQL", mysqlDorisDB, mysqlDorisTB, "site");

        DataStreamSink<RowData> rowDataDataStreamSink = dataStream.addSink(SinkBuilder.JdbcSink(mysqlDorisStr,
                (statement, event) -> {
                    String index = String.format("%s[%s]", indexDorisField, event.getString(0));
                    statement.setString(1, StatementUtil.subValue(index, event, 0, 38));
                    statement.setString(2, StatementUtil.subValue(index, event, 1, 38));
                    statement.setString(3, StatementUtil.subValue(index, event, 2, 30));
                    statement.setString(4, StatementUtil.subValue(index, event, 3, 512));
                    statement.setString(5, StatementUtil.subValue(index, event, 4, 40));
                    try {
                        Date opened = dateFormat.parse(StatementUtil.subValue(index, event, 5, 40));
                        java.sql.Date open = new java.sql.Date(opened.getTime());
                        statement.setDate(6, open);
                        Date closed = dateFormat.parse(StatementUtil.subValue(index, event, 6, 40));
                        java.sql.Date close = new java.sql.Date(closed.getTime());
                        statement.setDate(7, close);
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                    statement.setString(8, StatementUtil.subValue(index, event, 7, 40));
                    statement.setString(9, StatementUtil.subValue(index, event, 8, 40));
                    statement.setString(10, StatementUtil.subValue(index, event, 9, 40));

                    String mttr = StatementUtil.subValue(index, event, 10, 40);
                    statement.setBigDecimal(11, new BigDecimal(mttr));
//                    statement.setString(11, StatementUtil.subValue(index, event, 10, 40));
                    statement.setString(12, StatementUtil.subValue(index, event, 11, 40));
                    statement.setString(13, StatementUtil.subValue(index, event, 12, 40));
                    statement.setString(14, StatementUtil.subValue(index, event, 13, 40));
                    statement.setString(15, StatementUtil.subValue(index, event, 14, 40));
                    statement.setString(16, StatementUtil.subValue(index, event, 15, 40));
                    statement.setString(17, StatementUtil.subValue(index, event, 16, 512));
                    statement.setString(18, StatementUtil.subValue(index, event, 17, 40));
                    statement.setString(19, StatementUtil.subValue(index, event, 18, 512));
                    statement.setString(20, StatementUtil.subValue(index, event, 19, 512));
                    statement.setString(21, StatementUtil.subValue(index, event, 20, 100));
                }
                , mysqlDorisURL, MySQLConstant.DORIS_PRD02_USER_JABIL, MySQLConstant.DORIS_PRD02_USER_JABIL_PWD));

        env.execute(ExcelP1ToKpiJob.class.getSimpleName() + "_总数:" + rowData.size());

    }


    private static List<P1Entity> readExcel(String path) {
        List<P1Entity> p1list = new ArrayList<P1Entity>();
        try {
            FileInputStream fileInputStream = new FileInputStream(new File(path));
            Workbook workbook = new XSSFWorkbook(fileInputStream);
            Sheet sheetAt = workbook.getSheet("P1P2_For PBI");
            int rows = sheetAt.getLastRowNum();
            for (int i = 1; i <= rows; i++) {
                Row row = sheetAt.getRow(i);
                String c0 = getCellValueAsString(row.getCell(0));
                String c1 = getCellValueAsString(row.getCell(1));
                String c2 = getCellValueAsString(row.getCell(2));
                String c3 = getCellValueAsString(row.getCell(3));
                String c4 = getCellValueAsString(row.getCell(4));
                String c5 = getCellValueAsString(row.getCell(5));
                String c6 = getCellValueAsString(row.getCell(6));
                String c12 = getCellValueAsString(row.getCell(11));
                String c13 = getCellValueAsString(row.getCell(12));
                String c14 = getCellValueAsString(row.getCell(13));
                String c15 = getCellValueAsString(row.getCell(14));
                String c16 = getCellValueAsString(row.getCell(15));
                String c17 = getCellValueAsString(row.getCell(16));
                String c18 = getCellValueAsString(row.getCell(17));
                String c19 = getCellValueAsString(row.getCell(18));
                String c20 = getCellValueAsString(row.getCell(19));
                String c21 = getCellValueAsString(row.getCell(20));
                if (StrUtil.isNotBlank(c2) && StrUtil.isNotBlank(c0) && c0.trim().startsWith("1")) { //todo:只读取A列为1-的数据
                    Date cell5 = row.getCell(5).getDateCellValue();
                    Date cell6 = row.getCell(6).getDateCellValue();
                    long diff = cell6.getTime() - cell5.getTime();
                    long min = diff / (60 * 1000);
                    System.out.println("min: " + min);
                    String c8 = String.valueOf(min);
                    String c9 = getCellValueAsString(row.getCell(8));
                    String c10 =min >240 ? "TRUE" : "FALSE";
                    String c11 =   String.valueOf( new BigDecimal(min).setScale(2, RoundingMode.HALF_UP));

                    P1Entity p1Entity = new P1Entity(c0, c1, c2, c3, c4, c5, c6, c8, c9, c10, c11, c12, c13, c14, c15, c16, c17, c18, c19, c20, c21);
                    p1list.add(p1Entity);
                }
            }
        } catch (FileNotFoundException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return p1list;
    }

    private static String getCellValueAsString(Cell cell) {
        if (cell == null) {
            return "";
        }
        CellType cellType = cell.getCellType();
        String cellValue = "";
        switch (cellType) {
            // 如果数据类型错误，就不读了
            case ERROR:
                break;
            // 如果是字符串类型用getStringCellValue()取值
            case STRING:
                cellValue = cell.getStringCellValue();
                break;
            // 如果是布尔类型用getBooleanCellValue()取值
            case BOOLEAN:
                cellValue = String.valueOf(cell.getBooleanCellValue());
                break;
            // 如果是空，就不读了
            case BLANK:
                break;
            // 如果是数字类型，(日期、普通数字)
            case NUMERIC:
                if (HSSFDateUtil.isCellDateFormatted(cell)) {
                    // 如果是日期类型
                    Date date = cell.getDateCellValue();
                    cellValue = new DateTime(date).toString("yyyy-MM-dd");
                } else {
                    // 如果是普通数字
                    cellValue = cell.toString();
                }
                break;
        }
        return cellValue;
    }

    public static List<RowData> listToRowDataConverter(List<P1Entity> p1Entities) throws Exception {
        List<RowData> rrList = new ArrayList<>();
        SimpleDateFormat dateFormat = new SimpleDateFormat("dd-MM-yy");
        for (P1Entity entity : p1Entities) {
            RowKind rowKind = RowKind.INSERT;
            GenericRowData row = new GenericRowData(rowKind, 21);
            // 创建 RowData 对象并填充数据
            //Priority	Location	Number	Issue Description	Reason category	Opened	Closed	Mitunes	Week	Has Breached ? (SLA)	MTTR	Emergency Change
            // PRB Number	PRB Status	Actual RC Found (Yes/No)	Configuration Item	Impact	Production Line DT (Hour) 	Action Taken	Root Cause	Problem Solution
            row.setField(0, StringData.fromString(entity.getPriority()));//
            row.setField(1, StringData.fromString(entity.getLocation()));//
            row.setField(2, StringData.fromString(entity.getNumber()));//
            row.setField(3, StringData.fromString(entity.getIssue_description()));//
            row.setField(4, StringData.fromString(entity.getReason_category()));//
            row.setField(5, StringData.fromString(entity.getOpened()));//
            row.setField(6, StringData.fromString(entity.getClosed()));//
            row.setField(7, StringData.fromString(entity.getMitunes()));//
            row.setField(8, StringData.fromString(entity.getWeek()));//
            row.setField(9, StringData.fromString(entity.getHas_breached()));//
            row.setField(10, StringData.fromString(entity.getMttr()));//
            row.setField(11, StringData.fromString(entity.getEmergency_change()));//
            row.setField(12, StringData.fromString(entity.getPrb_number()));//
            row.setField(13, StringData.fromString(entity.getPrb_status()));
            row.setField(14, StringData.fromString(entity.getActual_rc_found()));
            row.setField(15, StringData.fromString(entity.getConfiguration_item()));
            row.setField(16, StringData.fromString(entity.getImpact()));
            row.setField(17, StringData.fromString(entity.getProduction_line_dt()));
            row.setField(18, StringData.fromString(entity.getAction_taken()));
            row.setField(19, StringData.fromString(entity.getRoot_cause()));
            row.setField(20, StringData.fromString(entity.getProblem_solution()));
            rrList.add(row);
        }
        return rrList;
    }
}
